Skip to main content

ML Training Thread

Operational Model​

Unlike the message-driven consumer threads, the ML Training Thread operates on a temporal schedule. It periodically queries the accumulated database for sufficient training data, performs model training when conditions are met, and publishes results for monitoring and deployment.

Training Data Management​

The thread implements sophisticated data sampling strategies through the get_training_data() method. For time series models like autoencoders, it creates sliding windows from sequential sensor readings, performs data quality checks, and applies normalization techniques. The AutoencoderMLTrainingThread demonstrates this by creating fixed-size windows from state sequences and converting them to training matrices.

Model Lifecycle Management​

Training follows a complete lifecycle: model architecture definition, data preprocessing, training execution, evaluation, and persistence. The thread maintains model versioning through the filesystem, stores training metadata alongside model weights, and implements atomic saves to prevent corruption during updates.

Adaptive Training Logic​

The system avoids unnecessary retraining by tracking data volume changes and training frequency. Training only occurs when sufficient new data has accumulated, preventing computational waste while ensuring models stay current with evolving data patterns.

Result Publishing​

Training outcomes, including loss metrics, evaluation results, and model metadata, are published to Kafka topics for system-wide visibility. This enables monitoring dashboards, automated model deployment decisions, and debugging capabilities.

User Implementation Requirements​

Four Required Methods​

def build_model(self):
    # Initialize model architecture (called during __init__)

def get_training_data(self) -> Optional[Any]:
    # Query database, check data sufficiency, return training data or None

def train_model(self, training_data: Any) -> Dict[str, Any]:
    # Execute training, return metrics dict

def eval_model(self) -> Dict[str, Any]:
    # Evaluate trained model, return evaluation metrics

def save_model(self, model_metrics: Dict, eval_results: Dict):
    # Persist model and metadata to filesystem/database

Implementation Steps​

  1. Build Model: Define architecture, initialize weights, set hyperparameters
  2. Check Data: Query self.db_manager for sufficient training samples
  3. Prepare Data: Create windows/batches, normalize, validate
  4. Train: Execute training loop, track metrics
  5. Evaluate: Test model performance, calculate thresholds
  6. Save: Store model files with versioning and metadata

Configuration Needed​

config = {
    'window_size': 50,
    'min_training_samples': 10000,
    'learning_rate': 0.001,
    'batch_size': 32,
    'epochs': 50,
    'kafka_topics': {'training_output': 'training-results'}
}

Environment Variables Required​

  • KAFKA_BROKER_URL
  • Database connection variables (same as data ingest)

What's Handled Automatically​

  • Kafka producer setup for publishing training results
  • Database connection management
  • Training loop timing and scheduling
  • Result publishing to Kafka topics
  • Thread lifecycle management
  • Error recovery and restart logic

Key Constraints​

  • Timer-based: Runs on schedule, not message-driven
  • Data checking: Only train when sufficient new data available
  • Model persistence: Handle atomic saves to prevent corruption
  • Memory management: Clean up training data after use
  • Blocking operations: Training can be long-running (single-threaded is acceptable)

The thread operates independently, checking for training opportunities and publishing results without external coordination.

Architectual Loop Diagram​

SMOCS ML Training Thread Diagram